package cn.com.klun.tjy.test.ZookeeperTest;

import org.apache.zookeeper.*;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;

public class ZkWorker implements Watcher{
    private static Logger logger = LoggerFactory.getLogger(ZkWorker.class);

    ZooKeeper zk;
    String hostPort;
    String status;
    String name = "woeker";
    String serverId = Integer.toHexString(new Random().nextInt());

    ZkWorker(String hostPort){
        this.hostPort = hostPort;
    }
    @Override
    public void process(WatchedEvent watchedEvent) {
        logger.info(watchedEvent.toString()+","+hostPort);
    }
    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort,15000,this);
    }
    void register(){
        zk.create("/workers/worker-"+serverId,
                "Idle".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL,
                createWorkerCallback,null);
    }

    StringCallback createWorkerCallback =new StringCallback() {
        @Override
        public void processResult(int i, String s, Object o, String s1) {
            switch (Code.get(i)){
                case CONNECTIONLOSS:
                    register();
                    break;
                case OK:
                    logger.info("创建节点成功"+serverId);
                    break;
                case NODEEXISTS:
                    logger.info("已经存在此节点"+serverId);
                    break;
                default:
                    logger.info("系统异常"+KeeperException.create(Code.get(i),s));
            }
        }
    };
    StatCallback statUpdateCallback = new StatCallback() {
        @Override
        public void processResult(int i, String s, Object o, Stat stat) {
            switch(Code.get(i)){
                case CONNECTIONLOSS:
                    updateStatus((String)o);
                    return;
            }

        }
    };
    synchronized private void updateStatus(String status){
        if(status == this.status){
            zk.setData("/workers/"+name,status.getBytes(),-1,statUpdateCallback,status);// -1表示不进行版本检查 无条件更新
        }
    }
    public void setStatus(String status){
        this.status = status;
        updateStatus(status);
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        ZkWorker w = new ZkWorker("192.168.137.12:2181");
        w.startZK();
        w.register();
        Thread.sleep(3000);
    }
}
